package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"os"
)

func main() {

	signals := make(chan os.Signal)

	// 1: 创建一个消费者
	c, _ := rocketmq.NewPushConsumer(
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"121.199.66.66:9876"})),
		consumer.WithGroupName("msg-group"),
		consumer.WithConsumerModel(consumer.BroadCasting),
		consumer.WithRetry(2),
	)

	topic := "order_broadcast_topic"
	// 2: 开始订阅消息
	err := c.Subscribe(topic, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			fmt.Printf("message id: %s, body: %s\n", msg.MsgId, string(msg.Body))
		}
		// 返回成功
		return consumer.ConsumeSuccess, nil
	})

	if err != nil {
		fmt.Println(err.Error())
	}

	err = c.Start()
	if err != nil {
		fmt.Println(err.Error())
		os.Exit(-1)
	}
	<-signals
	err = c.Shutdown()
	if err != nil {
		fmt.Printf("关闭消费者服务失败，%s", err.Error())
	}
}
